package com.hrt.kuducode

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.kudu.client._
import org.apache.kudu.client.KuduClient.KuduClientBuilder

/**
 * Flink操作kudu
 */
object FlinkKudu {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val ds: DataStream[String] = env.socketTextStream("cm3",9999)

    //自定义Sink - KuduSink
    ds.addSink(new RichSinkFunction[String] {
      var client: KuduClient = _
      var session: KuduSession = _

      //初始化调用一次，一般用作资源初始化
      override def open(parameters: Configuration): Unit ={
        client = new KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build()
        session = client.newSession()
      }

      //当sink数据时，调用
      override def invoke(line: String, context: SinkFunction.Context[_]): Unit = {
        //1,zs,18,100
        val arr: Array[String] = line.split(",")
        val id: Int = arr(0).toInt
        val name: String = arr(1)
        val age: Int = arr(2).toInt
        val score: Double = arr(3).toDouble

        val flink_t: KuduTable = client.openTable("flink_t1")

        val insert: Insert = flink_t.newInsert()
        val row: PartialRow = insert.getRow
        row.addInt("id",id)
        row.addString("name",name)
        row.addInt("age",age)
        row.addDouble("score",score)
        session.apply(insert)

      }

      //当flink 任务取消时，调用一次
      override def close(): Unit = {
        session.close()
        client.close()
      }


    })


    env.execute()
  }

}
